package com.study.chapter11;

import com.study.entity.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/22 21:05
 */
public class TabAndStreamTransformTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建实体表
        String createDDL = " CREATE TABLE click_table (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT " +
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'input/cart.txt', " +    //指定文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";

        tableEnv.executeSql(createDDL);
        Table clickTable = tableEnv.from("click_table");
        //注册虚拟表
        tableEnv.createTemporaryView("clickTable",clickTable);


        //将Table转换成流打印输出

        //1.调用toDataStream
        Table allTab = tableEnv.sqlQuery("select * from clickTable");
        tableEnv.toDataStream(allTab).print("allTab");

        //2.调用toChangelogStream()：聚合转换
        Table result = tableEnv.sqlQuery("select user,count(url) as num from clickTable group by user");
        tableEnv.toChangelogStream(result).print("result");


        DataStreamSource<Event> stream = env.fromElements(
                new Event("Bob", "/home", 10000L),
                new Event("Alice", "/home", 20000L),
                new Event("Jack", "/home", 10000L)
        );

        //将Stream转换成Table

        //1.调用fromDataStream
        Table table = tableEnv.fromDataStream(stream);
        //提取user、url为表中的列，且给user取别名为user_name，
        Table table1 = tableEnv.fromDataStream(stream, $("user").as("user_name"), $("url"));

        //2.调用createTemporaryView
        tableEnv.createTemporaryView("user_tab",stream, $("user").as("user_name"), $("url"));
        Table table2 = tableEnv.sqlQuery("select user_name,url from user_tab");


        env.execute();
    }
}
